package com.hub.realtime.flinkshims.core;

import com.hub.realtime.common.utils.RequireUtil;
import com.hub.realtime.flinkshims.core.enums.SqlErrorType;
import com.hub.realtime.flinkshims.core.model.SqlError;
import com.streamxhub.streamx.common.util.SqlSplitter;
import org.apache.commons.lang3.StringUtils;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.regex.Matcher;

/**
 * flink sql解析器
 */
public class SqlCommandParser {

    public static List<SqlCommandCall> parseSQL(String sql) {

        SqlError sqlEmptyError = new SqlError(SqlErrorType.VERIFY_FAILED, "sql is empty", sql);
        RequireUtil.require(StringUtils.isNotEmpty(sql) && StringUtils.isNotEmpty(sql.trim()), sqlEmptyError.toString());
        List<String> lines = new ArrayList<>();
        scala.collection.immutable.List<String> stringList = SqlSplitter.splitSql(sql);
        if (!stringList.isEmpty()) {
            for (int i = 0; i < stringList.length(); i++) {
                lines.add(stringList.apply(i));
            }
        }
        if (lines.isEmpty()) {
            throw new IllegalArgumentException(sqlEmptyError.toString());
        } else {
            List<SqlCommandCall> calls = new ArrayList<>();
            lines.forEach(stmt -> {
                Optional<SqlCommandCall> sqlCommandCall = parseLine(stmt);
                if (sqlCommandCall.isPresent()) {
                    calls.add(sqlCommandCall.get());
                } else {
                    throw new IllegalArgumentException((new SqlError(SqlErrorType.UNSUPPORTED_SQL, "unsupported sql", stmt)).toString());
                }
            });
            if (calls == null) {
                throw new IllegalArgumentException((new SqlError(SqlErrorType.SYNTAX_ERROR, "no executable sql", "")).toString());
            } else {

                return calls;
            }
        }
    }

    private static Optional<SqlCommandCall> parseLine(String sqlLine) {
        String stmt = sqlLine.trim();
        for (SqlCommand sqlCommand : SqlCommand.values()) {
            final Matcher matcher = sqlCommand.getPattern().matcher(stmt);
            if (matcher.matches()) {

                final String[] groups = new String[matcher.groupCount()];
                for (int i = 0; i < groups.length; i++) {
                    groups[i] = matcher.group(i + 1);
                }
                return sqlCommand.getOperandConverter().apply(groups)
                        .map((operands) -> new SqlCommandCall(sqlCommand, operands, sqlLine));

            }


        }
        return Optional.empty();
    }

}
